import com.zuikaku.pojo.Order;
import com.zuikaku.source.OrderSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

public class AggregateDemo {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironment();
        //2.设置模式AUTO，指定并行数为1方便观察
        environment.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        environment.setParallelism(1);
        //3.指定source源
        DataStreamSource<Order> orderDs = environment.addSource(new OrderSource());
        //4.对itemName进行分组
        KeyedStream<Order, String> keyedDs = orderDs.keyBy(new KeySelector<Order, String>() {
            @Override
            public String getKey(Order order) throws Exception {
                return order.getItemName();
            }
        });
        //5.开时间滚动窗口，window size = window slide =5,每5s统计过去5s的订单数据总额
        DataStream<Order> aggregated = keyedDs.window(TumblingProcessingTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
                //<IN,ACC,OUT>
                .aggregate(new AggregateFunction<Order, Order, Order>() {
                    //初始化累加器accumulator
                    @Override
                    public Order createAccumulator() {
                        Order accumulator =new Order();
                        //由于要求和，因此初始化好价格字段即可
                        accumulator.setPrice(new BigDecimal("0"));

                        return accumulator;
                    }
                    //定义聚合（广义累加）的逻辑
                    @Override
                    public Order add(Order in, Order accumulator) {

                        accumulator.setItemName(in.getItemName());
                        accumulator.setOrderId(in.getOrderId());
                        accumulator.setCustomerId(in.getCustomerId());
                        accumulator.setCreateDate(in.getCreateDate());
                        accumulator.setPrice(accumulator.getPrice().add(in.getPrice()));

                        //返回的对象，将作为下一次传入的累加器
                        return accumulator;
                    }

                    //获得结果
                    @Override
                    public Order getResult(Order accumulator) {
                        //结果是放在累加器的，可对累加器做额外业务操作，直接返回即可

                        return accumulator;
                    }

                    //合并结果（一般不需要）
                    @Override
                    public Order merge(Order order, Order acc1) {
                        return null;
                    }
                });
        aggregated.print("增量聚合后：");

        try {
            environment.execute("aggregate demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }
}
